package main;

import change.Revision;
import commonutils.MyCollectionUtils;
import commonutils.MyWikiObjectUtils;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowCallbackHandler;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallbackWithoutResult;
import org.springframework.transaction.support.TransactionTemplate;
import page.Lang;
import page.Link;
import page.WikiPage;
import pageblock.PageBlock;
import pageblock.PageBlockType;

import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.*;
import java.util.logging.Logger;

/**
 * Created by IntelliJ IDEA.
 * User: Evgeniy Krasko
 * Date: 06.11.11
 * Time: 19:15
 */
public class DatabaseChangesDepot implements ChangesDepot {
    private Logger logger = Logger.getLogger(DatabaseChangesDepot.class.getName());

    private final TransactionTemplate transactionTemplate;
    private final JdbcTemplate jdbcTemplate;

    public static final int PAGE_BATCH = 300;

    public DatabaseChangesDepot() {
        final BeanFactory beanFactory = new ClassPathXmlApplicationContext("config/dbconfig.xml");
        this.transactionTemplate = (TransactionTemplate) beanFactory.getBean("transactionTemplate");
        this.jdbcTemplate = (JdbcTemplate) beanFactory.getBean("jdbcTemplate");
    }


    public void submit(final List<PageBlock> changes) {
        final List<WikiPage> pages = MyWikiObjectUtils.getPagesByInfo(changes);
        final Map<WikiPage, Long> pageToId = getIdsByPages(pages);
        insertChanges(changes, pageToId);
    }

    public List<PageBlock> getChanges(final List<WikiPage> pages) {
        final List<PageBlock> result = new ArrayList<PageBlock>();
        for (List<WikiPage> pagesPart : MyCollectionUtils.splitList(pages, PAGE_BATCH)) {
            Map<WikiPage, Long> pagesToIds = getIdsByPages(pagesPart);

            StringBuilder sql = new StringBuilder("select * from changes where page_id in (");
            for (int i = 0; i < pagesToIds.values().size(); ++i) {
                sql.append((i == 0) ? "?" : ",?");
            }
            sql.append(");");

            final Map<Long, WikiPage> idsToPages = MyCollectionUtils.inverseBijectiveMap(pagesToIds);

            jdbcTemplate.query(sql.toString(),
                    pagesToIds.values().toArray(new Long[pagesToIds.values().size()]),
                    new RowCallbackHandler() {
                        public void processRow(ResultSet resultSet) throws SQLException {
                            result.add(new PageBlock(
                                    PageBlockType.getById(resultSet.getInt("elem_type")),
                                    new Revision(resultSet.getLong("rev_id"),
                                            idsToPages.get(resultSet.getLong("page_id")),
                                            new Date(resultSet.getDate("date").getTime())),
                                    resultSet.getInt("hash_code")));
                        }
                    });
        }
        return result;
    }

    public void resolveInterlinks(final List<Link> links) {
        final List<WikiPage> pages = MyWikiObjectUtils.getPagesByLinks(links);
        getIdsByPages(pages); // saves non-existing pages
        final Map<WikiPage, Long> pagesToEntityIds = getPagesWithExistingIds(pages, true);
        final Map<WikiPage, Set<WikiPage>> components = MyWikiObjectUtils.getComponents(links);
        for (Set<WikiPage> component : components.values()) {
            final WikiPage representative = component.iterator().next();
            final StringBuilder sql = new StringBuilder("update pages set entity_id = ? where entity_id in (");
            final List<Long> args = new ArrayList<Long>();
            args.add(pagesToEntityIds.get(representative));
            boolean first = true;
            for (WikiPage elem : component) {
                args.add(pagesToEntityIds.get(elem));
                sql.append((first) ? ("?") : (",?"));
                if (first) {
                    first = false;
                }
            }
            sql.append(");");
            jdbcTemplate.update(sql.toString(), args.toArray());
        }
    }

    private void insertChanges(Collection<PageBlock> changes, Map<WikiPage, Long> pageToId) {
        final String updateSql = "update changes set rev_id = ?, date = ?, hash_code = ? where (elem_type = ? and page_id = ?)";
        final List<Object[]> batchArgs = new LinkedList<Object[]>();

        for (final PageBlock change : changes) {
            final Object[] args = {
                    change.getRevision().getRevId(),
                    change.getRevision().getDate(),
                    change.getContentHashCode(),
                    change.getType().getTypeId(),
                    pageToId.get(change.getRevision().getWikiPage()),
            };
            batchArgs.add(args);
        }

        final int[] updateResult = jdbcTemplate.batchUpdate(updateSql, batchArgs);

        final String insertSql = "INSERT INTO changes (rev_id, date, hash_code, elem_type, page_id) VALUES (?,?,?,?,?)";

        final ListIterator<Object[]> batchArgsIterator = batchArgs.listIterator();
        for (int index : updateResult) {
            batchArgsIterator.next();
            if (index != 0) {
                batchArgsIterator.remove();
            }
        }
        jdbcTemplate.batchUpdate(insertSql, batchArgs);
    }

    private Map<WikiPage, Long> getIdsByPages(final List<WikiPage> pages) {
        final Map<WikiPage, Long> oldPages = getPagesWithExistingIds(pages, false);
        int nonExistingIdsCount = pages.size() - oldPages.size();
        long[] newIds = generateIds(nonExistingIdsCount);
        final Map<WikiPage, Long> newPages = new HashMap<WikiPage, Long>();
        int ptr = 0;
        for (final WikiPage page : pages) {
            if (!oldPages.containsKey(page)) {
                newPages.put(page, newIds[ptr++]);
            }
        }
        insertPages(newPages);
        final Map<WikiPage, Long> result = new HashMap<WikiPage, Long>();
        result.putAll(oldPages);
        result.putAll(newPages);
        return result;
    }

    private void insertPages(Map<WikiPage, Long> newPages) {
        final String sql = "INSERT INTO pages (title, lang, entity_id, id) VALUES (?,?,?,?)";
        final long[] entityIds = generateIds(newPages.size());
        int ptr = 0;
        final List<Object[]> batchArgs = new ArrayList<Object[]>();
        for (final WikiPage page : newPages.keySet()) {
            final Object[] args = {page.getTitle(), page.getLang().getIndex(), entityIds[ptr++], newPages.get(page)};
            batchArgs.add(args);
        }
        jdbcTemplate.batchUpdate(sql, batchArgs);
    }

    private Map<WikiPage, Long> getPagesWithExistingIds(List<WikiPage> pages, final boolean entityIds) {
        final Map<WikiPage, Long> result = new HashMap<WikiPage, Long>();
        for (List<WikiPage> pagesPart : MyCollectionUtils.splitList(pages, PAGE_BATCH)) {
            final StringBuilder sql = new StringBuilder("SELECT * FROM pages WHERE title in (");
            final Object[] args = new Object[pagesPart.size()];
            for (int i = 0; i < pagesPart.size(); ++i) {
                args[i] = pagesPart.get(i).getTitle();
                sql.append(i == 0 ? "?" : ",?");
            }
            sql.append(");");

            final Set<WikiPage> pagesSet = MyCollectionUtils.asSet(pages);

            jdbcTemplate.query(sql.toString(), args, new RowCallbackHandler() {
                public void processRow(ResultSet resultSet) throws SQLException {
                    final WikiPage curPage = new WikiPage(
                            Lang.getLang(resultSet.getInt("lang")),
                            resultSet.getString("title"));
                    final Long curId = resultSet.getLong(entityIds ? "entity_id" : "id");
                    if (pagesSet.contains(curPage)) {
                        result.put(curPage, curId);
                    }
                }
            });
        }
        return result;
    }

    public long[] generateIds(final int count) {
        if (count <= 0) {
            return new long[0];
        }
        final long[] ids = new long[count];
        transactionTemplate.execute(new TransactionCallbackWithoutResult() {
            @Override
            protected void doInTransactionWithoutResult(TransactionStatus transactionStatus) {
                ids[0] = jdbcTemplate.queryForLong("SELECT last_id FROM ids");
                jdbcTemplate.update("UPDATE ids SET last_id=?", ids[0] + count);
                logger.fine("Generating " + count + " id's.");
            }
        });
        for (int i = 1; i < count; ++i) {
            ids[i] = ids[0] + i;
        }
        return ids;
    }
}
